import org.apache.spark.{SparkConf, SparkContext}

object ab {
  def main(args: Array[String]): Unit = {
    // 1. 初始化Spark配置和上下文
    val conf = new SparkConf()
      .setAppName("SalaryRank")
      .setMaster("local[*]")  // 本地模式运行，使用所有CPU核心
    val sc = new SparkContext(conf)

    // 2. 读取本地CSV文件（需替换为实际路径）
    val filePath = "file:///D:/Employee_salary_first_half.csv"  // 本地路径格式参考‌:ml-citation{ref="1,7" data="citationList"}
    val rawRDD = sc.textFile(filePath)

    // 3. 删除标题行（首行）
    val header = rawRDD.first()
    val dataRDD = rawRDD.filter(_ != header)  // 过滤标题行‌:ml-citation{ref="1,3" data="citationList"}

    // 4. 提取姓名和薪资字段，并转换类型
    val processedRDD = dataRDD.map { line =>
      val columns = line.split(",")  // 按逗号分隔字段‌:ml-citation{ref="1" data="citationList"}
      val name = columns(1)          // 第2列为姓名（索引从0开始）
      val salary = columns(6).toInt  // 第7列为实际薪资，转为Int‌:ml-citation{ref="1,6" data="citationList"}
      (name, salary)
    }

    // 5. 按薪资降序排序
    val sortedRDD = processedRDD.sortBy(_._2, ascending = false)  // 降序排序‌:ml-citation{ref="1" data="citationList"}

    // 6. 获取前3名结果并输出
    val top3 = sortedRDD.take(3)  // 提取前3条记录‌:ml-citation{ref="1" data="citationList"}
    println("上半年实际薪资前三名员工：")
    top3.foreach { case (name, salary) =>
      println(s"姓名：$name\t薪资：$salary")
    }

    // 7. 关闭Spark上下文
    sc.stop()
  }
}